package cn.chihanbao.hbase.processor;

import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

public class FriendProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取表对象
        HTableWrapper friends = (HTableWrapper)e.getEnvironment().getTable( TableName.valueOf( "friend" ) );
        //获取到所有的cell对象
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        //通过队列集合
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        //循环队列获取cells对象
        for (Map.Entry<byte[], List<Cell>> entry : entries) {
            System.out.println(Bytes.toString(entry.getKey()));
            List<Cell> cells = entry.getValue();
            for (Cell cell : cells) {
                //rowkey信息
                byte[] rowkey = CellUtil.cloneRow( cell );
                //列信息
                byte[] column = CellUtil.cloneQualifier( cell );

                //验证删除的目标数据是否存在，存在则执行删除否则不执行,必须有此判断否则造成协处理器被循环调用耗尽资源
                boolean flag = friends.exists( new Get( column ).addFamily( Bytes.toBytes( "friends" ) ) );
                if (flag) {
                    Delete myDelete = new Delete( column ).addColumn( Bytes.toBytes( "friends" ), rowkey );
                    friends.delete( myDelete );
                }
            }
        }
    }
}
